package com.atguigu.datastream.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * ClassName: Flink03_Stream_WordCount_unBounded
 * Package: com.atguigu.day01
 * Description:
 *
 * @Author ChenJun
 * @Create 2023/4/6 18:34
 * @Version 1.0
 */
public class Flink03_Stream_WordCount_unBounded {
    public static void main(String[] args) throws Exception {
        // 1. 创建Flink 流运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(1);

        //2. 从端口读取文件（无界流）
        DataStreamSource<String> streamSource = env.socketTextStream("hadoop102", 9999);

        //3.将内容拆分为一个个的单词
        SingleOutputStreamOperator<String> wordStream = streamSource.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception {
                String[] words = s.split(" ");
                for (String word : words) {
                    collector.collect(word);
                }
            }
        });


        //4.将每一个单词组成二元组
        SingleOutputStreamOperator<Tuple2<String, Long>> wordToOne = wordStream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String s) throws Exception {
                return Tuple2.of(s, 1L);
            }
        });

        //5. 通过keyBy进行分组合并
        KeyedStream<Tuple2<String, Long>, Tuple> keyBy = wordToOne.keyBy(0);

        //6.sum计算
        SingleOutputStreamOperator<Tuple2<String, Long>> result = keyBy.sum(1);

        //6. 打印
        result.print();

        //7. 执行
        env.execute();

    }
}
